package com.gitee.huanminabc.utils_tools.redis.redisTemplate.utils;

import com.gitee.huanminabc.utils_tools.redis.redisTemplate.config.RedisStreamConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Redis Stream 消息处理器
 * 
 * 专门处理 Redis Stream 的消息发送、监听、重试等功能
 * 
 * @author sby
 */
@Component("sbyPluginRedisStreamMessageHandler")
@Slf4j
@ConditionalOnBean(RedisTemplate.class)
public class RedisStreamMessageHandler {

    /**
     * redisTemplate
     */
    @Autowired
    private RedisTemplate<String, Object> cacheRedisTemplate;

    /**
     * Redis Stream 配置
     */
    @Autowired(required = false)
    private RedisStreamConfig streamConfig;

    /**
     * Stream监听器容器
     */
    private volatile StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamContainer;

    /**
     * 活跃的订阅列表（使用线程安全的Map）
     */
    private final Map<String, Subscription> activeSubscriptions = new ConcurrentHashMap<>();

    /**
     * 消息重试次数记录（使用线程安全的Map）
     */
    private final Map<String, Integer> messageRetryCount = new ConcurrentHashMap<>();

    /**
     * 重试记录时间戳（使用线程安全的Map）
     */
    private final Map<String, Long> retryTimestamps = new ConcurrentHashMap<>();

    /**
     * 定时任务执行器（用于清理过期重试记录）
     */
    private final ScheduledExecutorService cleanupScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
        Thread t = new Thread(r, "redis-stream-cleanup");
        t.setDaemon(true);
        return t;
    });

    /**
     * 是否已初始化
     */
    private final AtomicBoolean initialized = new AtomicBoolean(false);

    /**
     * 是否已关闭
     */
    private final AtomicBoolean shutdown = new AtomicBoolean(false);

    /**
     * 清理任务失败次数
     */
    private final AtomicBoolean cleanupTaskStarted = new AtomicBoolean(false);

    /**
     * 默认配置值
     */
    private static final int DEFAULT_MAX_RETRY_COUNT = 3;
    private static final int DEFAULT_RETRY_RECORD_EXPIRE_HOURS = 24;
    private static final int DEFAULT_CLEANUP_INTERVAL_HOURS = 1;
    private static final boolean DEFAULT_ENABLE_AUTO_CLEANUP = true;

    /**
     * 初始化配置
     */
    @PostConstruct
    public void init() {
        // 验证必要的依赖
        Assert.notNull(cacheRedisTemplate, "cacheRedisTemplate未初始化");
        
        // 如果配置类未注入，使用默认配置
        if (streamConfig == null) {
            log.warn("RedisStreamConfig未配置，使用默认配置");
            streamConfig = createDefaultConfig();
        }
        
        // 验证配置参数
        validateConfig();
        
        // 启动清理任务
        startCleanupScheduler();
        
        log.info("RedisStreamMessageHandler初始化完成，配置: maxRetryCount={}, retryRecordExpireHours={}, cleanupIntervalHours={}, enableAutoCleanup={}", 
                getMaxRetryCount(), getRetryRecordExpireHours(), getCleanupIntervalHours(), isEnableAutoCleanup());
    }

    /**
     * 创建默认配置
     */
    private RedisStreamConfig createDefaultConfig() {
        RedisStreamConfig config = new RedisStreamConfig();
        config.setMaxRetryCount(DEFAULT_MAX_RETRY_COUNT);
        config.setRetryRecordExpireHours(DEFAULT_RETRY_RECORD_EXPIRE_HOURS);
        config.setCleanupIntervalHours(DEFAULT_CLEANUP_INTERVAL_HOURS);
        config.setEnableAutoCleanup(DEFAULT_ENABLE_AUTO_CLEANUP);
        return config;
    }

    /**
     * 验证配置参数
     */
    private void validateConfig() {
        if (getMaxRetryCount() < 0) {
            throw new IllegalArgumentException("最大重试次数不能小于0");
        }
        if (getRetryRecordExpireHours() <= 0) {
            throw new IllegalArgumentException("重试记录过期时间必须大于0");
        }
        if (getCleanupIntervalHours() <= 0) {
            throw new IllegalArgumentException("清理间隔时间必须大于0");
        }
        if (getCleanupIntervalHours() > getRetryRecordExpireHours()) {
            log.warn("清理间隔时间({})大于重试记录过期时间({})，可能导致清理不及时", 
                    getCleanupIntervalHours(), getRetryRecordExpireHours());
        }
    }

    /**
     * 发送Stream消息
     * 
     * 使用示例：
     * <pre>
     * // 发送简单消息
     * Map<String, String> message = new HashMap<>();
     * message.put("user", "张三");
     * message.put("action", "login");
     * String messageId = streamHandler.sendMessage("user_events", message);
     * 
     * // 发送单字段消息
     * String messageId = streamHandler.sendMessage("notifications", "type", "email");
     * </pre>
     *
     * @param streamKey Stream的key
     * @param message   消息内容
     * @return 消息ID
     */
    public String sendMessage(String streamKey, Map<String, String> message) {
        // 参数验证
        Assert.hasText(streamKey, "streamKey不能为空");
        Assert.notNull(message, "message不能为空");
        Assert.notNull(cacheRedisTemplate, "cacheRedisTemplate未初始化");
        
        try {
            StreamOperations<String, String, String> streamOps = cacheRedisTemplate.opsForStream();
            return streamOps.add(streamKey, message).getValue();
        } catch (Exception e) {
            log.error("发送Stream消息失败 streamKey:{}, message:{}", streamKey, message, e);
            throw new RuntimeException("发送Stream消息失败", e);
        }
    }

    /**
     * 发送Stream消息（简化版本，使用单个字段）
     *
     * @param streamKey Stream的key
     * @param field     字段名
     * @param value     字段值
     * @return 消息ID
     */
    public String sendMessage(String streamKey, String field, String value) {
        // 参数验证
        Assert.hasText(streamKey, "streamKey不能为空");
        Assert.hasText(field, "field不能为空");
        Assert.notNull(value, "value不能为空");
        
        Map<String, String> message = new HashMap<>();
        message.put(field, value);
        return sendMessage(streamKey, message);
    }

    /**
     * 监听Stream消息
     * 
     * 使用示例：
     * <pre>
     * // 监听用户事件
     * streamHandler.listenMessage("user_events", "user_group", "consumer1", message -> {
     *     Map<String, String> value = message.getValue();
     *     String user = value.get("user");
     *     String action = value.get("action");
     *     System.out.println("用户 " + user + " 执行了 " + action + " 操作");
     *     
     *     // 如果处理成功，会自动提交（ack）
     *     // 如果抛出异常，消息会重新投递，最多重试指定次数
     * });
     * </pre>
     *
     * @param streamKey    Stream的key
     * @param groupName    消费者组名称
     * @param consumerName 消费者名称
     * @param callback     消息处理回调函数
     */
    public void listenMessage(String streamKey, String groupName, String consumerName, 
                             java.util.function.Consumer<MapRecord<String, String, String>> callback) {
        // 参数验证
        Assert.hasText(streamKey, "streamKey不能为空");
        Assert.hasText(groupName, "groupName不能为空");
        Assert.hasText(consumerName, "consumerName不能为空");
        Assert.notNull(callback, "callback不能为空");
        Assert.notNull(cacheRedisTemplate, "cacheRedisTemplate未初始化");
        
        // 检查是否已关闭
        if (shutdown.get()) {
            throw new IllegalStateException("RedisStreamMessageHandler已关闭");
        }
        
        try {
            // 确保Stream存在
            ensureStreamExists(streamKey);
            
            // 确保消费者组存在
            ensureConsumerGroupExists(streamKey, groupName);
            
            // 创建监听器容器（如果不存在）
            ensureStreamContainer();
            
            // 检查是否已存在相同的订阅
            String subscriptionKey = streamKey + ":" + groupName + ":" + consumerName;
            
            // 使用同步块确保订阅操作的原子性
            synchronized (this) {
                Subscription existingSubscription = activeSubscriptions.remove(subscriptionKey);
                if (existingSubscription != null) {
                    log.warn("已存在相同的订阅，先停止旧订阅 streamKey:{}, groupName:{}, consumerName:{}", 
                            streamKey, groupName, consumerName);
                    try {
                        existingSubscription.cancel();
                    } catch (Exception e) {
                        log.error("停止旧订阅时发生异常", e);
                    }
                }
                
                // 创建订阅
                Subscription subscription = streamContainer.receive(
                    Consumer.from(groupName, consumerName),
                    StreamOffset.create(streamKey, ReadOffset.lastConsumed()),
                    message -> {
                        String messageKey = streamKey + ":" + message.getId().getValue();
                        try {
                            // 执行回调函数
                            callback.accept(message);
                            // 如果没有异常，自动提交
                            cacheRedisTemplate.opsForStream().acknowledge(groupName, message);
                            // 清除重试记录
                            messageRetryCount.remove(messageKey);
                            retryTimestamps.remove(messageKey);
                            log.debug("Stream消息处理成功并已提交 streamKey:{}, messageId:{}", streamKey, message.getId().getValue());
                        } catch (Exception e) {
                            handleMessageProcessingError(message, streamKey, groupName, messageKey, e);
                        }
                    }
                );
                
                // 保存订阅
                activeSubscriptions.put(subscriptionKey, subscription);
            }
            log.info("Stream监听器启动成功 streamKey:{}, groupName:{}, consumerName:{}", streamKey, groupName, consumerName);
            
        } catch (Exception e) {
            log.error("启动Stream监听器失败 streamKey:{}, groupName:{}, consumerName:{}", streamKey, groupName, consumerName, e);
            throw new RuntimeException("启动Stream监听器失败", e);
        }
    }

    /**
     * 处理消息处理错误
     */
    private void handleMessageProcessingError(MapRecord<String, String, String> message, String streamKey, 
                                            String groupName, String messageKey, Exception e) {
        // 获取当前重试次数
        int retryCount = messageRetryCount.getOrDefault(messageKey, 0);
        int maxRetryCount = getMaxRetryCountInternal();
        
        if (retryCount >= maxRetryCount) {
            // 超过最大重试次数，强制提交并记录错误
            try {
                cacheRedisTemplate.opsForStream().acknowledge(groupName, message);
                messageRetryCount.remove(messageKey);
                retryTimestamps.remove(messageKey);
                log.error("Stream消息处理失败，已达到最大重试次数({})，强制提交 streamKey:{}, messageId:{}", 
                        maxRetryCount, streamKey, message.getId().getValue(), e);
            } catch (Exception ackException) {
                log.error("强制提交消息失败 streamKey:{}, messageId:{}", streamKey, message.getId().getValue(), ackException);
            }
        } else {
            // 增加重试次数
            messageRetryCount.put(messageKey, retryCount + 1);
            retryTimestamps.put(messageKey, System.currentTimeMillis());
            log.error("Stream消息处理失败，第{}次重试 streamKey:{}, messageId:{}", 
                    retryCount + 1, streamKey, message.getId().getValue(), e);
            // 处理失败时不提交，消息会重新投递
        }
    }

    /**
     * 确保Stream容器已创建
     */
    private void ensureStreamContainer() {
        if (streamContainer == null) {
            synchronized (this) {
                if (streamContainer == null) {
                    RedisConnectionFactory connectionFactory = cacheRedisTemplate.getConnectionFactory();
                    if (connectionFactory == null) {
                        throw new IllegalStateException("RedisConnectionFactory未初始化");
                    }
                    streamContainer = StreamMessageListenerContainer.create(connectionFactory);
                    streamContainer.start();
                    log.info("Stream监听器容器已创建并启动");
                }
            }
        }
    }

    /**
     * 检查是否已初始化
     *
     * @return 是否已初始化
     */
    public boolean isInitialized() {
        return initialized.get();
    }

    /**
     * 检查是否已关闭
     *
     * @return 是否已关闭
     */
    public boolean isShutdown() {
        return shutdown.get();
    }

    /**
     * 停止Stream监听器
     *
     * @param streamKey    Stream的key
     * @param groupName    消费者组名称
     * @param consumerName 消费者名称
     */
    public void stopListener(String streamKey, String groupName, String consumerName) {
        // 参数验证
        Assert.hasText(streamKey, "streamKey不能为空");
        Assert.hasText(groupName, "groupName不能为空");
        Assert.hasText(consumerName, "consumerName不能为空");
        
        String subscriptionKey = streamKey + ":" + groupName + ":" + consumerName;
        
        // 使用同步块确保操作的原子性
        synchronized (this) {
            Subscription subscription = activeSubscriptions.remove(subscriptionKey);
            if (subscription != null) {
                try {
                    subscription.cancel();
                    log.info("Stream监听器已停止 streamKey:{}, groupName:{}, consumerName:{}", streamKey, groupName, consumerName);
                } catch (Exception e) {
                    log.error("停止Stream监听器时发生异常 streamKey:{}, groupName:{}, consumerName:{}", 
                            streamKey, groupName, consumerName, e);
                }
            } else {
                log.warn("未找到要停止的监听器 streamKey:{}, groupName:{}, consumerName:{}", streamKey, groupName, consumerName);
            }
        }
    }

    /**
     * 停止所有Stream监听器
     */
    public void stopAllListeners() {
        // 停止所有订阅
        for (Map.Entry<String, Subscription> entry : activeSubscriptions.entrySet()) {
            try {
                entry.getValue().cancel();
            } catch (Exception e) {
                log.error("停止订阅时发生异常 key:{}", entry.getKey(), e);
            }
        }
        activeSubscriptions.clear();
        
        // 清除重试记录
        messageRetryCount.clear();
        retryTimestamps.clear();
        
        // 停止清理任务
        stopCleanupScheduler();
        
        // 停止Stream容器
        if (streamContainer != null) {
            try {
                streamContainer.stop();
                streamContainer = null;
                log.info("Stream监听器容器已停止");
            } catch (Exception e) {
                log.error("停止Stream监听器容器时发生异常", e);
            }
        }
        
        log.info("所有Stream监听器已停止");
    }

    /**
     * 配置最大重试次数
     * 
     * 注意：建议通过配置文件设置，而不是通过代码设置
     * 配置文件示例：
     * sby:
     *   redis:
     *     stream:
     *       max-retry-count: 5
     *
     * @param maxRetryCount 最大重试次数
     */
    public void setMaxRetryCount(int maxRetryCount) {
        if (maxRetryCount < 0) {
            throw new IllegalArgumentException("最大重试次数不能小于0");
        }
        if (streamConfig != null) {
            streamConfig.setMaxRetryCount(maxRetryCount);
            log.info("设置最大重试次数为: {}", maxRetryCount);
        } else {
            log.warn("streamConfig未初始化，无法设置最大重试次数");
        }
    }

    /**
     * 获取最大重试次数
     *
     * @return 最大重试次数
     */
    public int getMaxRetryCount() {
        return getMaxRetryCountInternal();
    }

    /**
     * 内部获取最大重试次数方法
     */
    private int getMaxRetryCountInternal() {
        return streamConfig != null ? streamConfig.getMaxRetryCount() : DEFAULT_MAX_RETRY_COUNT;
    }

    /**
     * 获取重试记录过期时间
     */
    private int getRetryRecordExpireHours() {
        return streamConfig != null ? streamConfig.getRetryRecordExpireHours() : DEFAULT_RETRY_RECORD_EXPIRE_HOURS;
    }

    /**
     * 获取清理间隔时间
     */
    private int getCleanupIntervalHours() {
        return streamConfig != null ? streamConfig.getCleanupIntervalHours() : DEFAULT_CLEANUP_INTERVAL_HOURS;
    }

    /**
     * 是否启用自动清理
     */
    private boolean isEnableAutoCleanup() {
        return streamConfig != null ? streamConfig.isEnableAutoCleanup() : DEFAULT_ENABLE_AUTO_CLEANUP;
    }

    /**
     * 确保Stream存在
     *
     * @param streamKey Stream的key
     */
    private void ensureStreamExists(String streamKey) {
        try {
            // 尝试添加一条测试消息，如果Stream不存在会自动创建
            Map<String, String> testMessage = new HashMap<>();
            testMessage.put("_init", "true");
            String messageId = cacheRedisTemplate.opsForStream().add(streamKey, testMessage).getValue();
            // 立即删除测试消息
            cacheRedisTemplate.opsForStream().delete(streamKey, messageId);
        } catch (Exception e) {
            log.warn("确保Stream存在时发生异常 streamKey:{}", streamKey, e);
        }
    }

    /**
     * 确保消费者组存在
     *
     * @param streamKey Stream的key
     * @param groupName 消费者组名称
     */
    private void ensureConsumerGroupExists(String streamKey, String groupName) {
        try {
            cacheRedisTemplate.opsForStream().createGroup(streamKey, groupName);
            log.debug("消费者组创建成功 streamKey:{}, groupName:{}", streamKey, groupName);
        } catch (Exception e) {
            // 如果消费者组已存在，会抛出异常，这是正常的
            log.debug("消费者组可能已存在 streamKey:{}, groupName:{}", streamKey, groupName);
        }
    }

    /**
     * 清理过期的重试记录
     */
    private void cleanupExpiredRetryRecords() {
        try {
            long now = System.currentTimeMillis();
            List<String> keysToRemove = new ArrayList<>();
            int expireHours = getRetryRecordExpireHours();
            
            for (Map.Entry<String, Long> entry : retryTimestamps.entrySet()) {
                if (now - entry.getValue() > TimeUnit.HOURS.toMillis(expireHours)) {
                    keysToRemove.add(entry.getKey());
                }
            }
            
            for (String key : keysToRemove) {
                messageRetryCount.remove(key);
                retryTimestamps.remove(key);
                log.debug("清理过期重试记录 key:{}", key);
            }
            
            if (!keysToRemove.isEmpty()) {
                log.info("清理了{}条过期重试记录", keysToRemove.size());
            }
            
            // 记录当前重试记录数量，用于监控
            int currentCount = messageRetryCount.size();
            if (currentCount > 1000) {
                log.warn("当前重试记录数量较多: {}，建议检查消息处理逻辑", currentCount);
            }
        } catch (Exception e) {
            log.error("清理过期重试记录时发生异常", e);
            // 不抛出异常，避免清理任务中断
            // 但记录错误，便于监控
        }
    }

    /**
     * 启动清理任务
     */
    private void startCleanupScheduler() {
        if (cleanupTaskStarted.getAndSet(true)) {
            log.warn("清理任务已经启动，跳过重复启动");
            return;
        }
        
        if (!isEnableAutoCleanup()) {
            log.info("自动清理功能已禁用");
            return;
        }
        
        try {
            int intervalHours = getCleanupIntervalHours();
            cleanupScheduler.scheduleWithFixedDelay(
                this::cleanupExpiredRetryRecords,
                intervalHours, // 初始延迟
                intervalHours, // 执行间隔
                TimeUnit.HOURS
            );
            log.info("重试记录清理任务已启动，每{}小时清理一次", intervalHours);
        } catch (Exception e) {
            log.error("启动清理任务失败", e);
            cleanupTaskStarted.set(false);
        }
    }

    /**
     * 停止清理任务
     */
    private void stopCleanupScheduler() {
        if (!cleanupScheduler.isShutdown()) {
            try {
                cleanupScheduler.shutdown();
                // 等待任务完成，最多等待5秒
                if (!cleanupScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    cleanupScheduler.shutdownNow();
                }
                log.info("重试记录清理任务已停止");
            } catch (InterruptedException e) {
                log.error("停止清理任务时被中断", e);
                Thread.currentThread().interrupt();
                cleanupScheduler.shutdownNow();
            }
        }
    }


    /**
     * 应用关闭时清理资源
     */
    @PreDestroy
    public void destroy() {
        if (shutdown.getAndSet(true)) {
            log.warn("RedisStreamMessageHandler已经关闭，跳过重复关闭");
            return;
        }
        
        log.info("开始关闭RedisStreamMessageHandler...");
        stopAllListeners();
        log.info("RedisStreamMessageHandler已关闭");
    }
}
